package net.bwie.realtime.newcar.dws.function;
import net.bwie.realtime.newcar.dws.job.NewCarBean;
import net.bwie.realtime.newcar.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Iterator;


public class CarChargingAnalysisFuction extends KeyedProcessFunction<String,NewCarBean,String>{
    // 存储充电状态
    private transient ListState<NewCarBean> carStatus;
    @Override
    public void open(Configuration parameters) throws Exception {
        carStatus = getRuntimeContext().getListState(new ListStateDescriptor<NewCarBean>("carNewStatus", NewCarBean.class));
    }

    @Override
    public void processElement(NewCarBean newCarBean,Context context, Collector<String> out) throws Exception {
        // 添加当前数据
        carStatus.add(newCarBean);
        // 获取所用充电数据
        Iterable<NewCarBean> newCarBeans = carStatus.get();
        ArrayList<NewCarBean> newCarBeansList = new ArrayList<>();
        for (NewCarBean newCarBean1 : newCarBeans) {
            newCarBeansList.add(newCarBean1);
        }
//        System.out.println("Current list size: " + newCarBeansList.size());
        // 是否完成充电
        boolean fullyCharged  = newCarBeansList.stream()
                .anyMatch(data -> data.getBatteryPower() >= 70.2);
//        System.out.println("Fully charged: " + fullyCharged);
        // 找到最早时间
        if(fullyCharged) {
            NewCarBean startData = newCarBeansList.get(0);
            for (NewCarBean newCarBean1 : newCarBeansList) {
                if (newCarBean1.getBatteryPower() < startData.getBatteryPower()) {
                    startData = newCarBean1;
                }
            }

            // 找到充电结束的数据
            NewCarBean endData = null;
            for (NewCarBean newCarBean1 : newCarBeansList) {
                if (newCarBean1.getBatteryPower() == 70.2 && (endData == null || newCarBean1.getTimestamp() > endData.getTimestamp())) {
                    endData = newCarBean1;
                }
            }
            // 计算结果
            if (endData != null) {
                //汽车id
                String vin = startData.getVin();
                //事件时间
                Long timestamp = startData.getTimestamp();
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                String format = sdf.format(timestamp);
                //充电总时长
                long chargingTime = endData.getTimestamp() - startData.getTimestamp();
                //充电开始电量
                Double startBatteryPower = startData.getBatteryPower();
                //充电完成电量
                Double endBatteryPower = endData.getBatteryPower();
                //充电总量
                double sumBatteryPower = endBatteryPower - startBatteryPower;
                //充电总量段
                String sumBatteryPowerLevel = "充电不足30%";
                long timeConsuming = 0L;
                if(sumBatteryPower >= 30.0 && startBatteryPower < 50.0){
                    sumBatteryPowerLevel = "30%";
                    timeConsuming = chargingTime;
                }else if(sumBatteryPower >= 50.0 && startBatteryPower < 80.0){
                    sumBatteryPowerLevel = "50%";
                    timeConsuming = chargingTime;
                }else if(sumBatteryPower >= 80.0 && startBatteryPower < 100.0){
                    sumBatteryPowerLevel = "80%";
                    timeConsuming = chargingTime;
                }else if(startBatteryPower >= 100.0){
                    sumBatteryPowerLevel = "100%";
                    timeConsuming = chargingTime;
                }
                String output = vin +","+ format +","+ chargingTime +","+ startBatteryPower +","+ endBatteryPower +","+ sumBatteryPower
                        +","+ sumBatteryPowerLevel +","+ timeConsuming;
                out.collect(output);
                //清理状态
                carStatus.clear();
            }
        }
    }
}